package com.shangguigu.hadoopdashujv.pojo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

public class HDFSOpearation {
    private FileSystem fs;
    //配置环境
    public HDFSOpearation() throws Exception{
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://node1:8020");

        //设置客户端身份
        System.setProperty("HADOOP_USER_NAME", "hadoop");


        fs = FileSystem.get(URI.create("hdfs://node1:8020"),conf);
    }

    public static void main(String[] args) throws Exception{
        HDFSOpearation ops = new HDFSOpearation();

//        ops.mkdir("/testjava01");

        //测试上传
//        ops.uploadFile("C:\\Users\\Administrator\\Desktop\\c.txt", "/testjava01/testupload.txt");
//        ops.uploadFile("/home/hadoop/test01/testupload.txt", "/testjava01/testupload2.txt");

//        ops.listFiles("/testjava01");
        //测试下载

//        ops.downloadFile("/testjava01/testupload.txt", "C:\\Users\\Administrator\\Desktop\\hadoop\\testdownload.txt");

        //测试合并
//        ops.mergeLocalFiles("C:\\Users\\Administrator\\Desktop\\hadoop\\", "/testjava01/testmerge.txt");

        //测试追加
//        ops.appendToFile("C:\\Users\\Administrator\\Desktop\\hadoop\\hebing.txt","/testjava01/testupload.txt");

        //先过滤出所有.txt文件
        Path hdfsPath = new Path("/testjava01");
        List<Path> txtfiles=new ArrayList<>();
        FileStatus[] statuses = ops.fs.listStatus(hdfsPath);
        for (FileStatus status : statuses) {
            if (status.getPath().getName().endsWith(".txt")) {
                txtfiles.add(status.getPath());
            }
        }

        //合并所有内容到merged.txt
        Path mergedPath = new Path("/testjava01/merged.txt");

        FSDataOutputStream out = ops.fs.create(mergedPath);
        for (Path txtfile : txtfiles) {
            InputStream in = ops.fs.open(txtfile);
            IOUtils.copyBytes(in, out, 4096, false);
            in.close();
        }
        out.close();
        System.out.println("文件合并成功: " + txtfiles + " -> " + mergedPath);



    }




    //创建目录
    public void mkdir(String path) throws Exception {
        Path hdfsPath = new Path(path);
        if (fs.exists(hdfsPath)) {
            System.out.println("目录已存在: " + path);
        } else {
            fs.mkdirs(hdfsPath);
            System.out.println("目录创建成功: " + path);
        }
    }
    //列出目录
    public void listFiles(String path) throws Exception {
        Path hdfsPath = new Path(path);
        FileStatus[] statuses = fs.listStatus(hdfsPath);



        System.out.println("目录 " + path + " 内容:");
        for (FileStatus status : statuses) {
            System.out.println((status.isDirectory() ? "目录" : "文件") + ": "
                    + status.getPath().getName()
                    + " 大小: " + status.getLen() + " bytes");
        }
    }
    //上传文件
    public void uploadFile(String localPath, String hdfsPath) throws Exception {
        Path local = new Path(localPath);
        Path hdfs = new Path(hdfsPath);

        //判断本地路径是否正确
        System.out.println(local.getFileSystem(new Configuration()).exists(local));
        if (!local.getFileSystem(new Configuration()).exists(local)) {
            System.out.println("本地文件不存在: " + localPath);
        }

        fs.copyFromLocalFile(local, hdfs);
        System.out.println("文件上传成功: " + localPath + " -> " + hdfsPath);
    }

    //下载文件
    public void downloadFile(String hdfsPath, String localPath) throws Exception {
        Path hdfs = new Path(hdfsPath);
        Path local = new Path(localPath);

        fs.copyToLocalFile(hdfs, local);
        System.out.println("文件下载成功: " + hdfsPath + " -> " + localPath);
    }

    //本地文件合并到hdfs
    public void mergeLocalFiles(String localDir, String hdfsFile) throws Exception {
        Path hdfsPath = new Path(hdfsFile);
        FSDataOutputStream out = fs.create(hdfsPath);

        File localDirFile = new File(localDir);
        File[] localFiles = localDirFile.listFiles();

        for (File file : localFiles) {
            if (file.isFile()) {
                InputStream in = new BufferedInputStream(new FileInputStream(file));
                IOUtils.copyBytes(in, out, 4096, false);
                in.close();
            }
        }
        out.close();
        System.out.println("本地文件合并完成: " + localDir + " -> " + hdfsFile);
    }

    //追加文件内容
    public void appendToFile(String localFile, String hdfsFile) throws Exception {
        Path hdfsPath = new Path(hdfsFile);
        FSDataOutputStream out = fs.append(hdfsPath);
        InputStream in = new BufferedInputStream(new FileInputStream(localFile));

        IOUtils.copyBytes(in, out, 4096, false);

        in.close();
        out.close();
        System.out.println("内容追加完成: " + localFile + " -> " + hdfsFile);
    }
}
